/*
 * SocketAcceptor.java    0.0.1    22/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.net.cs.server.impl;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.SocketAddress;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashSet;
import java.util.Set;

import org.mentalsmash.tazio.net.cs.CSSubsystemException;
import org.mentalsmash.tazio.net.cs.RuntimeCSSubsystemException;
import org.mentalsmash.tazio.net.cs.server.DeflatingProtocolHandler;
import org.mentalsmash.tazio.net.cs.server.ProtocolHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The SocketAcceptor is an implementation of {@link Acceptor Acceptor} that
 * runs in a thread of its own.
 * <p>
 * A <code>SocketAcceptor</code> is associated on a socket on which to listen
 * for incoming connections. The thread that runs the acceptor is started
 * calling the {@link SocketAcceptor#start() start()} method on it: it creates a
 * thread that listens for connections on the given socket and dispatches new
 * connection to the dispatcher associated to the acceptor.
 * </p>
 * <p>
 * The association Acceptor-Dispatcher <strong>has to be</strong> set before the
 * SocketAcceptor thread is started, and cannot be changed while it is running.
 * </p>
 * 
 * @version 0.0.1 22/set/2009
 * @author Andrea Reale <andrea.reale@gmail.com>
 */
public class SocketAcceptor implements DeflatingAcceptor, Runnable {

    private final static Logger log = LoggerFactory
	    .getLogger(SocketAcceptor.class);

    private final static String THREAD_NAME = "Acceptor";

    private final Set<AcceptorListener> registeredListeners; // the listeners
							     // registered on
							     // the acceptors.
    // Accesses to this set is synchronised using the set's lock.

    private volatile boolean running = false; // controls the listening cycle

    private Dispatcher dispatcher; // where to dispatch incoming connections
    private ServerSocketChannel incomingSocket; // the socket to listen for
						// connections on

    private Class<? extends ProtocolHandler> protocolHandlerClass; // The class
								   // that will
								   // be used to
    // instantiate protocol handler objects

    private Thread acceptorThread = null; // keeps a reference to the acceptor
					  // thread, if it is running;

    private final Object statusLock = new Object(); // Used to control accesses
						    // on private dispatcher
    // and protocol handler class

    private SocketAddress address; // used to trye to restore the sercer socket
				   // upon exceptions

    // true to have DeflatingProtocolHandler instantiated with the deflating
    // option set to true
    private boolean deflating = false;

    /**
     * Create a new <code>SocketAcceptor</code> and sets the specified values.
     * <p>
     * If the <code>dispatcher</code> needs to be started before dispatching,
     * you <strong>must</strong> call its <code>start()</code> method prior to
     * calling this constructor. <strong>Notice</strong> that if the dispatcher
     * is not in the <code>dispatching</code> no error is thrown, and only a
     * warning message is logged.
     * </p>
     * <p>
     * The passed <code>socketChan</code> needs to be open and its associated
     * socket already bound.
     * </p>
     * <p>
     * The <code>handlerClass</code> must be a ProtocolHandler implementation
     * and must provide a default constructor.
     * </p>
     * 
     * @param dispatcher
     *            the dispatcher that will be used by this acceptor to handle
     *            new connections. It <strong>must</strong> already have been
     *            started
     * @param handlerClass
     *            the class of the protocol handlers that are to be created to
     *            handle messages coming from connections. It must have a
     *            default constructor.
     * @param socketChan
     *            a {@link java.nio.channels.ServerSocketChannel
     *            ServerSocketChannel} to listen on. The channel
     *            <strong>must</strong> be open, and its associated socket
     *            <strong>must</strong> be bound to a valid address and port.
     * @throws IOException
     *             if the <code>socket</code> is not bound.
     * @throws CSSubsystemException
     *             If the dispatcher is not started.
     * 
     */
    public SocketAcceptor(Dispatcher dispatcher,
	    Class<? extends ProtocolHandler> handlerClass,
	    ServerSocketChannel socketChan) throws IOException,
	    CSSubsystemException {

	if (!socketChan.isOpen()) {
	    String msg = "SocketChannel passed as argument must be open";
	    log.error(msg);
	    throw new IOException(msg);
	}

	if (!socketChan.socket().isBound()) {
	    log
		    .error("The socket associated to the passed socket channel must be bound");
	    throw new IOException("The socket" + socketChan + " was not bound");
	}

	registeredListeners = new HashSet<AcceptorListener>();

	setDispatcher(dispatcher);

	protocolHandlerClass = handlerClass;

	incomingSocket = socketChan;
	address = socketChan.socket().getLocalSocketAddress();
    }

    /**
     * Create a new <code>SocketAcceptor</code> and sets the specified values.
     * <p>
     * If the <code>dispatcher</code> needs to be started before dispatching,
     * you <strong>must</strong> call its <code>start()</code> method prior to
     * calling this constructor.<strong>Notice</strong> that if the dispatcher
     * is not in the <code>dispatching</code> no error is thrown, and only a
     * warning message is logged.
     * </p>
     * <p>
     * The <code>handlerClass</code> must be a ProtocolHandler implementation
     * and must provide a default constructor.
     * </p>
     * 
     * @param dispatcher
     *            the dispatcher that will be used by this acceptor to handle
     *            new connections. It <strong>must</strong> already have been
     *            started.
     * @param handlerClass
     *            the class of the protocol handlers that are to be created to
     *            handle messages coming from connections. It must have a
     *            default constructor.
     * @param address
     *            a valid InetAddres: it will be used to bind a socket on.
     * @param port
     *            a port for the socket to listen on
     * @throws IOException
     *             if a socket on the given <code>address</code> and
     *             <code>port</code> cannot be bound.
     * @throws CSSubsystemException
     *             If the dispatcher is not started.
     */
    public SocketAcceptor(Dispatcher dispatcher,
	    Class<? extends ProtocolHandler> handlerClass, InetAddress address,
	    int port) throws IOException, CSSubsystemException {

	registeredListeners = new HashSet<AcceptorListener>();

	setDispatcher(dispatcher);

	incomingSocket = ServerSocketChannel.open();
	ServerSocket socket = incomingSocket.socket();

	this.address = new InetSocketAddress(address, port);
	socket.bind(this.address);

	protocolHandlerClass = handlerClass;
    }

    /**
     * Sets the class that will be used to instantiate {@link ProtocolHandler
     * ProtocolHandler} objects, which in turn are used to handle incoming data
     * from client connections.
     * 
     * @param protocolHandlerClass
     *            A class instance which implements {@link ProtocolHandler
     *            ProtocolHandler} . It has to provide a default constructor.
     */
    // CONCURRENCY: any thread could call this. Modifications to the protocol
    // handler, must be mutually exclusive
    // wrt creation of new ProtocolHandler instances.
    public void setProtocolHandlerClass(
	    Class<? extends ProtocolHandler> protocolHandlerClass) {
	synchronized (statusLock) {
	    this.protocolHandlerClass = protocolHandlerClass;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#getCurrentDispatcher()
     */
    // CONCURRENCY: any thread could call this. Accesses and Modifications to
    // the current dispatcher
    // have to be synchronized wrt registrations on the dispatcher
    @Override
    public Dispatcher getCurrentDispatcher() {
	synchronized (statusLock) {
	    return dispatcher;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.net.cs.server.Acceptor#setDispatcher(org.mentalsmash
     * .tazio.net.cs.server.Dispatcher)
     */
    @Override
    // CONCURRENCY: any thread could call this. Accesses and Modifications to
    // the current dispatcher
    // have to be synchronized wrt registrations on the dispatcher
    public void setDispatcher(Dispatcher dispatcher)
	    throws CSSubsystemException {

	if (isRunning()) {
	    String msg = "Cannot set the dispatcher for this acceptor: the acceptor is already running";
	    log.error(msg);
	    throw new CSSubsystemException(msg);
	}

	synchronized (statusLock) {
	    // FIXME: maybe I should have not commented this!!
	    // if(!dispatcher.isDispatching()) {
	    // throw new CSSubsystemException(new
	    // DispatcherNotRunningException());
	    // }
	    this.dispatcher = dispatcher;
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#isRunning()
     */
    // CONCURRENCY: running is volatile :)
    @Override
    public boolean isRunning() {
	return running;
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#start()
     */
    // CONCURRENCY: since only a thread at once can run this acceptor, this
    // method is synchronised
    // on the acceptor itself to avoid multiple threads start a new acceptor
    @Override
    public Thread start() {
	synchronized (this) {

	    if (running) {
		throw new RuntimeCSSubsystemException(
			"One Acceptor is already running");
	    }

	    acceptorThread = new Thread(this);
	    acceptorThread.setName(THREAD_NAME);

	    acceptorThread.start();
	    while (!running) {
		try {
		    log.debug("Waiting for acceptor thread to start");
		    wait(); // waiting for the acceptorThread to actually start
		} catch (InterruptedException e) {
		    log.error("Unexpected Interrupt received ", e);
		    continue;
		}
	    }
	}
	log.debug("Acceptor thread started");
	// notify the listeners
	notifyStart();

	return acceptorThread;
    }

    /**
     * Notifies the registered listeners that the acceptor has started. This
     * method is called by the thread that made the Dispatcher Thread start
     */
    private void notifyStart() {
	synchronized (registeredListeners) {
	    for (AcceptorListener l : registeredListeners) {
		l.acceptorStarted();
	    }
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#stop()
     */
    // CONCURRENCY: synchronised on the acceptor itself: not sure if that it is
    // strictly necessary
    // to avoid multiple threads call stop on the thread. But it won't be
    // harmful.
    @Override
    public synchronized void stop() {
	if (!running) {
	    log.debug("Cannot sto acceptor: it is not running.");
	    return;
	}

	log.debug("Acceptor was asked to stop, working for it...");

	running = false;

	if (acceptorThread != null && acceptorThread.isAlive()) {
	    acceptorThread.interrupt(); // if the serversocket is blocked on a
					// accept() raises the
	    // closed by interrupt exception and closes the socket
	    log.debug("Waiting for the " + THREAD_NAME + " to finish");
	    try {
		if (running) { // if it is still alive
		    acceptorThread.join();
		}
	    } catch (InterruptedException e) {
		// No - op: thread finished :)
	    }
	    log.debug("Acceptor thread stopped... cleaning up");

	    beforeStop(); // maybe it's redundant

	}

	// listeners are notified when the acceptor Thread lunches the
	// interrupted exception in the mothod run()

    }

    /*
     * (non-Javadoc)
     * 
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {

	try {
	    synchronized (statusLock) {
		while (!(incomingSocket != null && incomingSocket.isOpen() && incomingSocket
			.socket().isBound())) {
		    log.trace("Rebinding sockets");
		    incomingSocket = ServerSocketChannel.open();
		    incomingSocket.socket().bind(address);
		}
	    }
	    running = true;
	    synchronized (this) {
		notifyAll(); // notifies the thread that started the acceptor
			     // that the acceptor has started.
	    }
	    listenForConnections();
	} catch (ClosedByInterruptException e) {
	    log.warn("The " + THREAD_NAME
		    + " thread was closed by an external interrupt");
	    beforeStop();
	    notifyStop(e);
	    return;
	} catch (ClosedChannelException e) {
	    log.error("Error: ServerSocketChannel closed: " + e, e);
	    beforeStop();
	    notifyStop(e);
	    return;
	} catch (IOException e) {
	    log.error("Error: Unexpetcer IO error: " + e, e);
	    beforeStop();
	    notifyStop(e);
	    return;
	} catch (Throwable e) {
	    log.error("FATAL: unexpexted error: " + e, e);
	    beforeStop();
	    notifyStop(e);
	    return;
	}
    }

    /**
     * Notifies every listener that the dispatcher has stopped. This method is
     * called by the acceptor thread
     * 
     * @param e
     *            the cause of the stop
     */
    private void notifyStop(Throwable e) {
	synchronized (registeredListeners) {
	    for (AcceptorListener l : registeredListeners) {
		l.acceptorStopped(e);
	    }
	}

    }

    /**
     * Cleans status variables of this acceptor instance. Should be called after
     * server shutdown or unexpected fatal errors.
     */
    // CONCURRENCY: It won't be of any harm not to synchronise this method since
    // running is volatile,
    // and the rest of the operations are idempotent.
    private void beforeStop() {
	running = false;

	try {
	    synchronized (statusLock) {

		incomingSocket.close();
		incomingSocket = null;
	    }
	} catch (Exception e) {
	    // don't care, I'm quitting.
	}

    }

    /**
     * Implements the "listening-cycle"
     */
    // This method should be run only from the acceptor thread, that should be
    // one and only one for
    // acceptor instance. Anyway I need to synchronise accesses to other state
    // variables from other
    // threads
    private void listenForConnections() throws IOException {
	while (running) {
	    SocketChannel connection = incomingSocket.accept();
	    if (connection == null) {
		continue;
	    }
	    ProtocolHandler handler = null; // the handler for data coming in
					    // from the newly accepted
					    // connection

	    /*
	     * No one can change the dispatcher or the class for protocol
	     * handlers while the connection is being accepted and set-up on the
	     * dispatcher
	     */
	    synchronized (statusLock) {
		try {
		    handler = protocolHandlerClass.newInstance();
		    if (handler instanceof DeflatingProtocolHandler){
			((DeflatingProtocolHandler)handler).setDeflating(this.isDeflating());
		    }
		} catch (Exception e) {
		    String msg = "Unable to instantiate protocol handler: " + e;
		    log.warn(msg, e);
		}

		dispatcher.register(connection, handler);
	    }
	}
    }

    /**
     * Returns the thread that is currently running the acceptor
     * 
     * @return the thread that is currently running the acceptor
     */
    public synchronized Thread getAcceptorThread() {
	return this.acceptorThread;
    }

    /* ************************************* */
    /* Listener Registration related methods */
    /* ************************************* */

    @Override
    // CONCURRENCY: synchronised on the set itself
    public boolean registerListener(AcceptorListener listener) {
	synchronized (registeredListeners) {
	    return registeredListeners.add(listener);
	}
    }

    @Override
    // CONCURRENCY: synchronised on the set itself
    public boolean removeListener(AcceptorListener listener) {
	synchronized (registeredListeners) {
	    return registeredListeners.remove(listener);
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.net.cs.server.impl.DeflatingAcceptor#setDeflating
     * (boolean)
     */
    @Override
    public void setDeflating(boolean compressed) {
	this.deflating = compressed;
    }
    
    public boolean isDeflating() {
	return this.deflating;
    }

}
